package com.resoft.bigdata.p2p.service.api.thrift2imp;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.thrift2.generated.TColumn;
import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.TResult;
import org.apache.hadoop.hbase.thrift2.generated.TScan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.resoft.bigdata.p2p.record.DebtorInfo;
import com.resoft.bigdata.p2p.record.InvestInfo;
import com.resoft.bigdata.p2p.record.LoanContract;
import com.resoft.bigdata.p2p.record.MonthlyStatistics;
import com.resoft.bigdata.p2p.record.RecordInterface;
import com.resoft.bigdata.p2p.service.api.QueryService;

public class Thrift2QueryServiceImpl extends AbsThrift2Client implements QueryService{
	private static final Logger logger = LoggerFactory.getLogger(Thrift2QueryServiceImpl.class);
	public Thrift2QueryServiceImpl(String host, int port) {
		super(host, port);
	}

	@Override
	public List<String> getOrgCodes() throws IOException {
		return getOrgCodes(null);
	}
	
	@Override
	public List<String> getOrgCodes(String date) throws IOException {
		List<String> orgs = new ArrayList<String>();
		String table = LoanContract.TABLE_NAME;
		
		byte[] startRow = Bytes.toBytes(0);
		byte[] stopRow = Bytes.toBytes(-1);
		byte[] lastRow = null;
		
		try {
			transport.open();
		} catch (TTransportException e) {
			logger.error("Thrift2 Client open transport failed:", e.getMessage());
			e.printStackTrace();
			return orgs;
		}
		int totalNumber = 0;
		while(true){
			TScan scan = new TScan();
			if(lastRow != null){
				int orgCode = Bytes.toInt(lastRow, 0, 4);
				startRow = Bytes.toBytes(++orgCode);
			}
			scan.setStartRow(startRow);
			scan.setStopRow(stopRow);
			scan.setMaxVersions(1);
			List<TColumn> columns = new ArrayList<TColumn>();
			
			TColumn column = new TColumn();
			column.setFamily(Bytes.toBytes(LoanContract.FAMILY_NAMES[0]));
			column.setQualifier(LoanContract.ORG_CODE);
			columns.add(column);
			
			column = new TColumn();
			column.setFamily(Bytes.toBytes(LoanContract.FAMILY_NAMES[0]));
			column.setQualifier(LoanContract.DATE);
			columns.add(column);
			
			if(date != null && !date.isEmpty()){
				String filter = "ColumnPrefixFilter('DDATE') AND ValueFilter(=,'substring:"+date+"')";
				scan.setFilterString(filter.getBytes());
				System.out.println(" data is not null ....");
				
//				_Fields field = _Fields.findByName(new String(LoanContract.DATE));
//				scan.setFieldValue(field, date);
			}
			
			scan.setColumns(columns);
			int localNumber = 0;
			List<TResult> results;
			try {
				results = client.getScannerResults(ByteBuffer.wrap(table.getBytes()), scan, 1);
				for (TResult tResult : results) {
					++localNumber;
					lastRow = tResult.getRow();
					for (TColumnValue columnValue : tResult.getColumnValues()) {
						if(Bytes.compareTo(columnValue.getQualifier(), LoanContract.ORG_CODE) == 0)
						{
							orgs.add(new String(columnValue.getValue()));
						}
//						System.out.println(Bytes.toString(columnValue.getQualifier())+" = " +Bytes.toString(columnValue.getValue()));
					}
					
				}
	
				totalNumber += localNumber;
				if(localNumber == 0){
					System.out.println("BREAK.................");
					break;
				}
			} catch (TException e) {
				logger.error("Fetch Data failed! Message:" + e.getMessage());
				e.printStackTrace();
				transport.close();
				break;
			}
			
//			System.err.println("Total number:" + totalNumber + ", local number:" + localNumber +  ", row Key:" + Bytes.toHex(lastRow));
		} //~~while
	
		logger.debug("Get {} rows", totalNumber);
		transport.close();
		
		return orgs;
	}
	
	
	@Override
	public List<LoanContract> getLoanContractByOrgCodeAndDates(String orgCode,
			String beginDate, String endDate, long pageSize, byte[] lastRow)
			throws IOException {
		List<LoanContract> infos  = new ArrayList<LoanContract>();
		
		try {
			List<RecordInterface> records = getRecordsByOrgCodeAndDates(orgCode, beginDate, endDate, pageSize, lastRow, LoanContract.class);
			for(RecordInterface record: records){
				infos.add((LoanContract)record);
			}
		} catch (InstantiationException e) {
			logger.error("GetLoanContractByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			logger.error("GetLoanContractByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		}
		
		return infos;
	}
	
//	@Override
//	public List<LoanContract> getLoanContractByOrgCodeAndDates(String orgCode,
//			String beginDate, String endDate, long pageSize, byte[] lastRow)
//			throws IOException {
//		List<LoanContract> contracts = new ArrayList<LoanContract>();
//		String table = LoanContract.TABLE_NAME;
//		
//		String startRowDate = endDate;
//		String stopRowDate = DateUtil.addSomeDays(beginDate, -1);
//		byte[] startRow = Bytes.toBytes(0);
//		byte[] stopRow = createNextRowKeyForOrgCodeDate(orgCode, stopRowDate, lastRow);;
//		if(pageSize > DEFAULT_MAX_RESULT_SIZE || pageSize <1){
//			pageSize = DEFAULT_MAX_RESULT_SIZE;
//		}
//		
//		try {
//			transport.open();
//		} catch (TTransportException e) {
//			logger.error("Thrift2 Client open transport failed:", e.getMessage());
//			e.printStackTrace();
//			return contracts;
//		}
//		int totalNumber = 0;
////		LoanContract contractType = new LoanContract();
//		while(true){
//			TScan scan = new TScan();
//			startRow = createNextRowKeyForOrgCodeDate(orgCode, startRowDate, lastRow);
//			scan.setStartRow(startRow);
//			scan.setStopRow(stopRow);
//			scan.setMaxVersions(1);
//			
////			List<TColumn> columns = new ArrayList<TColumn>();
////			for(byte[]qualifier : contractType.getQualifiers()){
////				TColumn column = new TColumn();
////				column.setFamily(Bytes.toBytes(LoanContract.FAMILY_NAMES[0]));
////				column.setQualifier(qualifier);
////				columns.add(column);
////				
////			}
////			scan.setColumns(columns);
////			System.out.println("columns size:" + columns.size());
//
//			int localNumber = 0;
//			List<TResult> results;
//			try {
//				results = client.getScannerResults(ByteBuffer.wrap(table.getBytes()), scan, (int)pageSize);
//				for (TResult tResult : results) {
//					++localNumber;
//					lastRow = tResult.getRow();
//					LoanContract contract = new LoanContract();
//					contract.setStoredRowKey(lastRow);
//					for (TColumnValue columnValue : tResult.getColumnValues()) {
////						System.out.println(Bytes.toString(columnValue.getQualifier())+" = " +Bytes.toString(columnValue.getValue()));
//						contract.setValue(columnValue.getQualifier(), columnValue.getValue());
//					}
//					contracts.add(contract);
////					System.out.println();
//				}
//	
//				totalNumber += localNumber;
//				if(localNumber == 0){
//					System.out.println("BREAK.................");
//					break;
//				}
//			} catch (TException e) {
//				logger.error("Fetch Data failed! Message:" + e.getMessage());
//				e.printStackTrace();
//				transport.close();
//				break;
//			}
////			System.err.println("Total number:" + totalNumber + ", local number:" + localNumber +  ", row Key:" + Bytes.toHex(lastRow));
//		} //~~while
//
//		logger.debug("Get {} rows", totalNumber);
//		System.err.println("Total number:" + totalNumber );
//		
//		transport.close();
//		
//		return contracts;
//	}

	@Override
	public List<LoanContract> getLoanContractByOrgCodeAndOneDate(
			String orgCode, String date) throws IOException {
		List<LoanContract> allConstracts = new ArrayList<LoanContract>();
		byte[] lastRowKey = null;
		while(true){
			List<LoanContract> contracts  = getLoanContractByOrgCodeAndDates(orgCode,date, date, -1, lastRowKey);
			if(contracts != null && !contracts.isEmpty()){
				allConstracts.addAll(contracts);
				lastRowKey = contracts.get(contracts.size()-1).getStoredRowKey();
			}else{
				break;
			}
		}
		return allConstracts;
	}

	@Override
	public List<DebtorInfo> getDebtorInfoByOrgCodeAndDates(String orgCode,
			String beginDate, String endDate, long pageSize, byte[] lastRow) {
		List<DebtorInfo> infos  = new ArrayList<DebtorInfo>();
		
		try {
			List<RecordInterface> records = getRecordsByOrgCodeAndDates(orgCode, beginDate, endDate, pageSize, lastRow, DebtorInfo.class);
			for(RecordInterface record: records){
				infos.add((DebtorInfo)record);
			}
		} catch (InstantiationException e) {
			logger.error("GetDebtorInfoByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			logger.error("GetDebtorInfoByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		}
		
		return infos;
	}

	@Override
	public List<DebtorInfo> getDebtorInfoByOrgCodeAndOneDate(String orgCode,
			String date) {
		List<DebtorInfo> allDebtors = new ArrayList<DebtorInfo>();
		byte[] lastRowKey = null;
		while(true){
			List<DebtorInfo> debtors  = getDebtorInfoByOrgCodeAndDates(orgCode,date, date, -1, lastRowKey);
			if(debtors != null && !debtors.isEmpty()){
				allDebtors.addAll(debtors);
				lastRowKey = debtors.get(debtors.size()-1).getStoredRowKey();
			}else{
				break;
			}
		}
		return allDebtors;
	}

	@Override
	public List<InvestInfo> getInvestInfoByOrgCodeAndDates(String orgCode,
			String beginDate, String endDate, long pageSize, byte[] lastRow) {
		List<InvestInfo> infos  = new ArrayList<InvestInfo>();
		
		try {
			List<RecordInterface> records = getRecordsByOrgCodeAndDates(orgCode, beginDate, endDate, pageSize, lastRow, InvestInfo.class);
			for(RecordInterface record: records){
				infos.add((InvestInfo)record);
			}
		} catch (InstantiationException e) {
			logger.error("getInvestInfoByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			logger.error("getInvestInfoByOrgCodeAndDates failed, error message: " + e.getMessage());
			e.printStackTrace();
		}
		
		return infos;
	}

	@Override
	public List<InvestInfo> getInvestInfoByOrgCodeAndOneDate(String orgCode,
			String date) {
		List<InvestInfo> allInvestors = new ArrayList<InvestInfo>();
		byte[] lastRowKey = null;
		while(true){
			List<InvestInfo> investors  = getInvestInfoByOrgCodeAndDates(orgCode,date, date, -1, lastRowKey);
			if(investors != null && !investors.isEmpty()){
				allInvestors.addAll(investors);
				lastRowKey = investors.get(investors.size()-1).getStoredRowKey();
			}else{
				break;
			}
		}
		return allInvestors;
	}

	@Override
	public List<MonthlyStatistics> getAllp2pMonthlyStatistics(String beginDate,
			String endDate) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public List<MonthlyStatistics> getAllp2pMonthlyStatistics(String date) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public List<MonthlyStatistics> getp2pMonthlyStatistics(String orgCode,
			String beginDate, String endDate) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public List<MonthlyStatistics> getp2pMonthlyStatistics(String orgCode,
			String date) {
		// TODO Auto-generated method stub
		return null;
	}

}
